package com.karateca.columbiaimports;

import com.karateca.columbiaimports.http.HttpDownloader;
import com.karateca.columbiaimports.http.HttpDownloaderImpl;
import com.karateca.columbiaimports.io.ImportFile;
import com.karateca.columbiaimports.io.TextFileReader;
import java.io.File;
import java.net.URLEncoder;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * The download handler coordinates the sequence of steps to do the search according
 * to the start and end dates, get the links by country, download each import,
 * and write the imports to files. 
 * @author Andres Dominguez
 */
public class DownloadHandler
{
    /** time (in seconds) between statistics updates */
    public static final int STATISTICS_DELAY = 10;
    /** Sleep time between threads check */
    public static final int PING_DELAY = 1000;
    /** Number of active threads */
    private int threadCount;
    /** Directory where files will be downloaded */
    private String downloadDirectory;
    /** Apache log. */
    private static final Log log = LogFactory.getLog(DownloadHandler.class);
    /** List of countries that will not be downloaded */
    private List<String> skipCountryList;
    /** Start search date */
    private String startDate;
    /** End search date */
    private String endDate;
    /** File name with the thread count */
    public static final String THREAD_COUNT_FILE = "thread_count.txt";
    /** Maximum number of updates with zero download progress */
    public static final int MAX_ZERO_DELTA_COUNT = 50;
    /** Number of countries that are downloaded at a time */
    private int downloadBatchSize = 200;

    /**
     * Creates a new instance of DownloadHandler.
     * @param startDate start search date.
     * @param endDate end search date.
     * @param downloadDirectory directory where files will be written.
     * @param threadCount number of concurrent threads.
     * @param downloadBatchSize Number of countries that are downloaded at a time.
     * @param skipCountryList list of countries that will not be downloaded.
     */
    public DownloadHandler(String startDate, String endDate, String downloadDirectory,
            int threadCount, int downloadBatchSize, List<String> skipCountryList) throws
            Exception
    {
        this.startDate = startDate;
        this.endDate = endDate;
        this.downloadDirectory = downloadDirectory;
        this.threadCount = threadCount;
        this.downloadBatchSize = downloadBatchSize;
        this.skipCountryList = skipCountryList;
    }

    /**
     * Start downloading data.
     * @throws java.lang.Exception in case of error.
     */
    public void startDownload() throws Exception
    {
        ImportDownloader importDownloader;
        HttpDownloader downloader;
        List<ImportBean> importBeans;
        List<ImportBean> downloadBatch;
        Map<String, List<String>> customsCountryMap;
        ImportFile importFile;
        int i = 0;
        String startD;
        String endD;
        boolean skippedCountry;

        // Encode dates
        startD = URLEncoder.encode(this.startDate, "ISO-8859-1");
        endD = URLEncoder.encode(this.endDate, "ISO-8859-1");

        downloader = new HttpDownloaderImpl(6);
        importDownloader = new ImportDownloader(startD, endD, downloader);

        // Do the search and find the links by customs and countries
        log.info("Start searching countries for each customs");
        customsCountryMap = importDownloader.readCountryCustomsLinks();
        log.info("Done searching, found " + customsCountryMap.size() + " countries");

        // Remove all unwanted countries from the map
        customsCountryMap.keySet().removeAll(this.skipCountryList);

        // For each country, get the import links, download the data, and write
        // the results to a file
        for (String country : customsCountryMap.keySet())
        {
            skippedCountry = false;
            log.info("Downloading country [" + country + "] [" + ++i + " of " + customsCountryMap.size() + "]");
            // Get all the links
            importBeans = importDownloader.readCountryImportLinks(customsCountryMap.get(country));
            // if a temporary file exists then remove the links already written
            importFile = new ImportFile(this.downloadDirectory, country, startDate, endDate);
            log.info("[" + country + "] Found " + importBeans.size() + " links");
            if (importFile.tempFileExists())
            {
                log.info("[" + country + "] found existing temp file");
                removeCompletedLinks(importFile, importBeans);
            }

            // Download and write imports in batches
            downloadBatch = new LinkedList<ImportBean>();
            while (importBeans.size() > 0)
            {
                log.info("[" + country + "] remaining links: " + importBeans.size() + " links");
                // Copy beans to batch
                for (int j = 0; j < downloadBatchSize; j++)
                {
                    if (importBeans.size() == 0)
                    {
                        break;
                    }
                    downloadBatch.add(importBeans.remove(0));
                }

                // Download data and check result
                if (doCountry(country, downloadBatch, downloader))
                {
                    // Write data
                    writeFile(country, downloadBatch);
                }
                else
                {
                    log.error("[" + country + "] Could not download country, skip to next");
                    skippedCountry = true;
                    break;
                }
                // Free memory
                downloadBatch.clear();
            }
            if (skippedCountry == false)
            {
                // rename temp file to complete file
                log.info("[" + country + "] Done writing all imports, changing file name");
                importFile.markComplete();
            }
        }
        log.info("Download complete");
    }

    /**
     * Remove all the imports that have been downloaded already. These imports
     * are in the temp file.
     * @param importFile temp file containing the imports that have been downloaded
     * already.
     * @param importBeans list of import found in the search.
     * @throws Exception in case of error.
     */
    private void removeCompletedLinks(ImportFile importFile, List<ImportBean> importBeans) throws Exception
    {
        Set<String> tempFileLinks;
        String link;
        int beansRemoved = 0;

        log.info("Removing existing links");
        tempFileLinks = importFile.readTempFileLinks();
        // Remove all the beans that are already in the temp file
        while (importBeans.size() > 0)
        {
            link = importBeans.get(0).getLink();
            if (tempFileLinks.contains(link))
            {
                tempFileLinks.remove(link);
                importBeans.remove(0);
                beansRemoved++;
            }
            else
            {
                break;
            }
        }
        // If the links in the temp file are not in the list there is an error.
        if (tempFileLinks.size() != 0)
        {
            String errMsg = "Some of the links in the temp file are not in search results";

            log.error(errMsg);
            Iterator<String> it = tempFileLinks.iterator();
            while (it.hasNext())
            {
                log.error("Missing link: " + it.next());
            }
            throw new Exception(errMsg);
        }
        log.info("Removed " + beansRemoved + " existing links");
    }

    /**
     * Write import file for a country.
     * @param country country code.
     * @param importBeans beans with import data.
     * @throws java.lang.Exception when the file cannot be opened.
     */
    private void writeFile(String country, List<ImportBean> importBeans) throws Exception
    {
        ImportFile importFile;
        boolean fileExists;

        importFile = new ImportFile(this.downloadDirectory, country, startDate, endDate);
        log.info("Writing file to " + importFile.getTempFileName());
        fileExists = importFile.tempFileExists();
        importFile.openTempFile(fileExists);

        // if file is new write columns
        if (fileExists == false)
        {
            importFile.writeColumnNames();
        }
        // Write each bean
        for (ImportBean bean : importBeans)
        {
            try
            {
                importFile.writeBean(bean, country);
            }
            catch (Exception e)
            {
                log.error("Error writing import bean with link " + bean.getLink(), e);
                bean.printBean();
            }
        }
        importFile.closeFile();
    }

    /**
     * Download the data for a country.
     * @param country country code.
     * @param links import beans with the links to be downloaded.
     * @param downloader initialized downloader.
     * @return true when download is successful. false otherwise.
     */
    private boolean doCountry(String country, List<ImportBean> links, HttpDownloader downloader) throws InterruptedException
    {
        // TODO remove threads working method
        List<ImportBean> downloadQueue;
        DownloadWorker worker;
        DownloadStatistics statistics;
        int pingCount = 0;
        ThreadGroup threadGroup;

        // Start statistics
        statistics = new DownloadStatistics(links.size());
        log.info("[" + country + "] batch has " + links.size() + " links");
        // Create synchronized queue and start threads
        downloadQueue = Collections.synchronizedList(new ArrayList<ImportBean>());
        downloadQueue.addAll(links);
        worker = new DownloadWorker(downloadQueue, downloader, log);
        // Start threads
        threadGroup = new ThreadGroup("Worker thred group");
        startWorkerThreads(threadGroup, worker, downloadQueue.size());

        // Keep reporting progress until all threads are done working
        while (worker.getThreadsWorking() > 0)
        {
            // Check if any of the threads crashed
            if (worker.isThreadCrashed())
            {
                // kill threads and exit loop
                worker.setMaxThreads(0);
                break;
            }

            pingCount++;
            if ((pingCount % STATISTICS_DELAY) == 0)
            {
                // Update statistics, the number equals imports remaining plus items being downloaded
                statistics.updateCount(downloadQueue.size() + worker.getThreadsWorking());
                log.info("[" + country + "] " + statistics.getStatistics() + ", threads: " + worker.getThreadsWorking());
                // Check thread count file
                if (readThreadCountFile())
                {
                    // Thread file changed, update thread count and start new threads
                    worker.setMaxThreads(this.threadCount);
                    startWorkerThreads(threadGroup, worker, downloadQueue.size());
                }
                // Check if there if no progress
                if (statistics.getZeroDeltaCount() > MAX_ZERO_DELTA_COUNT)
                {
                    // Threads are hanging too long. Interrupt threads, put
                    // beans back in the queue and start a new set of threads
                    log.error("[" + country + "] Threads are hanging, creating new threads");
                    statistics.setZeroDeltaCount(0);
                    threadGroup.interrupt();
                    // wait for all threads to be done and put unfinished beans back in queue
                    interruptThreads(worker, threadGroup);
                    while (worker.getWorkingQueue().size() != 0)
                    {
                        downloadQueue.add(worker.getWorkingQueue().remove(0));
                    }
                    // Start threads again
                    startWorkerThreads(threadGroup, worker, downloadQueue.size());
                }
            }
            Thread.sleep(PING_DELAY);
        }

        // Check if any threads crashed
        if (worker.isThreadCrashed())
        {
            log.fatal("One or more threads crashed downloading country [" + country + "]");
            // wait for threads to finish
            log.info("Waiting one minute for threads to finish");
            worker.setMaxThreads(0);
            interruptThreads(worker, threadGroup);
            return false;
        }
        log.info("[" + country + "] Done downloading, time: " + statistics.getRunningTime());
        return true;
    }

    /**
     * Interrupts the threads that are currently downloading imports. It tries to
     * interrupt the threads for a minute. After that minute it gives up.
     * @param worker object downloading imports.
     * @param threadGroup thread group with the worker threads.
     * @throws InterruptedException when the sleep is interrupted.
     */
    private void interruptThreads(DownloadWorker worker, ThreadGroup threadGroup) throws
            InterruptedException
    {
        int j = 0;
        while (worker.getThreadsWorking() != 0)
        {
            j++;
            log.info("Obj count: " + worker.getThreadsWorking());
            log.info("Thread group count: " + threadGroup.activeCount());
            Thread.sleep(10000);
            // Give up after one minute
            if (j > 6)
            {
                log.error("Could not stop threads");
                System.exit(0);
            }
        }
    }

    /**
     * Start the threads in the worker. The number of threads that will be started
     * depends on the maxmim threads allowed in threadCount, the queue size, and
     * the threads that are currently running in the worker.
     * @see DownloadHandler#threadCount
     * @param threadGroup new threads will be added to this group.
     * @param worker worker object where new threads will run.
     * @param queueSize download queue size.
     */
    private void startWorkerThreads(ThreadGroup threadGroup, DownloadWorker worker, int queueSize)
    {
        // Number of threads that will be started
        int newThreadCount;
        // Number of threads currently working
        int threadsWorking;

        // Update maximum number of threads
        worker.setMaxThreads(this.threadCount);

        // Calculate how many threads have to be started
        threadsWorking = worker.getThreadsWorking();
        newThreadCount = Math.min((this.threadCount - threadsWorking), queueSize);
        for (int i = 1; i <= newThreadCount; i++)
        {
            log.info("Starting thread " + i);
            new Thread(threadGroup, worker, "DownloadThread" + i).start();
        }

        // Wait briefly for the threads to start
        try
        {
            Thread.sleep(PING_DELAY);
        }
        catch (InterruptedException e)
        {
            log.error("Got an interrupted exception", e);
        }
    }

    /**
     * The thread count can be updated in a file specified in THREAD_COUNT_FILE
     * in the download directory. The file contains an integer number.
     * @see #THREAD_COUNT_FILE
     * @return true if the thread count changed. false if it didn't change or
     */
    public boolean readThreadCountFile()
    {
        File threadFile;
        boolean valueChanged = false;
        int newThreadCount;
        String fileContents;

        try
        {
            // Check if file exists
            threadFile = new File(this.downloadDirectory + THREAD_COUNT_FILE);
            if (threadFile.exists())
            {
                // Read value
                fileContents = TextFileReader.readFile(this.downloadDirectory + THREAD_COUNT_FILE);
                // Check if value changed
                newThreadCount = Integer.parseInt(fileContents.trim());
                if (newThreadCount != this.threadCount)
                {
                    log.info("Thread file changed, new value: " + newThreadCount);
                    this.threadCount = newThreadCount;
                    valueChanged = true;
                }
            }
        }
        catch (Exception e)
        {
            log.error("Error reading thread file" + e.getMessage());
        }

        return valueChanged;
    }
}
